package com.learning.rocketmq.optimal.throughput;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * ClassName: SkipMessage
 * Description: 检测延时情况，跳过非重要消息
 * <p>
 * Consumer 在消费的过程中， 如果发现由于某种原因发生严重的消息堆积，短时间无法消除堆积，
 * 这个时候可以选择丢弃不重要的消息，使Consumer 尽快追上Producer 的进度
 * <p>
 * Date: 2019/1/15 9:05 【需求编号】
 *
 * @author Sam Sho
 * @version V1.0.0
 */
public class SkipMessage {


    public static void consumer() throws MQClientException {
        DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer("base_consumer_group");
        pushConsumer.setNamesrvAddr("10.0.64.106:9876;10.0.64.107:9876");
        pushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        // 用SQL 表达式的方式进行过滤
        pushConsumer.subscribe("property_topic", MessageSelector.bySql("a between 0 and 3"));


        pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                long offset = msgs.get(0).getQueueOffset();
                String maxOffset = msgs.get(0).getProperty(MessageConst.PROPERTY_MAX_OFFSET);
                long diff = Long.parseLong(maxOffset) - offset;

                // 当某个队列的消息数堆积到90000 条以上, 就直接丢弃
                if (diff > 90000) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }

                // 正常消费消息
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        pushConsumer.start();
    }
}
